package com.watchme.parser;

import com.watchme.parser.utils.PdfParserUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.BatchUpdate;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.File;
import java.io.IOException;

import static com.watchme.parser.PdfReader.base64StringToPDF;

/**
 * Created by Mengxy on 2017/11/23.
 */
public class ParsePdfHbase {
    private static HBaseConfiguration hbaseConfig=null;
    static{
        Configuration config=new Configuration();
        config.set("hbase.zookeeper.quorum","192.168.10.149,192.168.10.44,192.168.10.49");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        hbaseConfig=new HBaseConfiguration(config);
    }

    /**
     * get方式，通过rowKey查询
     * @param tablename
     * @param rowKey
     * @throws IOException
     */
    public static void selectByRowKey(String tablename,String rowKey) throws IOException {
        HTable table=new HTable(hbaseConfig,tablename);
        Get g = new Get(Bytes.toBytes(rowKey));
        Result r=table.get(g);
        for(KeyValue kv:r.raw()){
            System.out.println("column: "+new String(kv.getColumn()));
            System.out.println("value: "+new String(kv.getValue()));


            //向Hbase中插入数据
            BatchUpdate update=new BatchUpdate("Huangyi");
            update.put("name:java","http://www.sun.com".getBytes());
            table.commit(update);




        }
    }

    //扫描hbase表，获取所有记录
    public void getAllRow(HBaseConfiguration conf, String tableName){
        HTable hTable = null;
        ResultScanner rs = null;
        try {
            hTable = new HTable(conf, "DANIU");
            rs = hTable.getScanner(new Scan());
            //循环rowkey
            for (Result result : rs) {
                for (KeyValue kv : result.raw()) {
                    System.err.println("rowkey:" + new String(kv.getKey()));
                    System.err.println("-------------------------------");
                    System.err.println(
                            "columnFamily:" + new String(kv.getFamily()) +
                                    "===column:" + new String(kv.getQualifier()) +
                                    "===getValue:" + new String(kv.getValue()));

                    //解析转成而二进制的PDF文件
                    //1、将二进制转成PDF文件
                    String url = this.getClass().getResource("").getPath()+"//temp.pdf";
                    base64StringToPDF(new String(kv.getValue()),url);
                    //2、解析PDF文件
                   PdfParserUtil.getInstance().parser(url,"");
                    //3.删除PDF
                    File tempFile = new File(url);
                    boolean b = true;
                    //判断文件是否存在
                    if (tempFile.exists()) {
                        b = tempFile.delete();
                    }
                    if (b) {
                        System.out.println("删除文件成功");
                    } else {
                        System.out.println("删除文件出错");
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally{
            rs.close();
            try {
                hTable.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
